#include "fsm.h"

//=============================================================

void *wait_all_retnull(void *parptr, uint32_t timeout)
{
	Pieces *parallel;
	struct timeval tv;
	uint32_t i;
	void *retval = NULL;

	long int diff = 0L;

	if ((parallel = (Pieces *) parptr) == NULL)
		return (retval);
    
	do {

		gettimeofday(&tv, NULL);

		diff = (tv.tv_usec + 1000000 * tv.tv_sec) - (parallel->tv.tv_usec + 1000000 * parallel->tv.tv_sec);

		if (diff > (long int)timeout * 1000000) {
			while ( pthread_mutex_trylock(&parallel->lock) != 0 ) ;;
			for (i = 0; i < parallel->thread_count; i++) {
				pthread_cancel(parallel->threads[i]);
			}
			pthread_mutex_unlock(&parallel->lock);
			break;
		} 

	    while (	pthread_mutex_trylock(&parallel->lock) != 0 );;
		while(parallel->pcount != 0) {
			pthread_cond_wait(&parallel->cwait,&parallel->lock);
		}
		pthread_mutex_unlock(&parallel->lock);

	} while (0);


	return (retval);

}


void *boost_running(void *xFilter, list *glist, void *dst_addr, Boosting funcs, Shrunken sk_func, uint32_t timeout)
{

	Pieces *parallel=NULL;
	uint32_t i, groups, stride, last, Continue = 0;
	void *tmplist = NULL;
    void **node_vector_ptr;

	Pbricks *bptr;

    pthread_attr_t attr;
    pthread_mutexattr_t mutexattr;
    int rc;

	do {
		if (funcs == NULL) break;
		if (glist == NULL) break;
		if (listLength(glist) == 0) break;
		if ((parallel = Calloc(1,Pieces)) == NULL) break;

		pthread_mutex_init(&parallel->lock, NULL);
		pthread_cond_init(&parallel->cwait,NULL);

		parallel->pcount = 0;

		last = parallel->len_of_mark = listLength(glist);

		parallel->shrunken = sk_func;
		groups = 2 * core_count();

		do {
			stride = (parallel->len_of_mark + groups - 1) / groups;
			if (stride < 2 ) groups = (groups + 1) / 2;
		} while ((stride < 2) && (groups > 1));

		parallel->thread_count = groups;

		gettimeofday(&parallel->tv, NULL);

        node_vector_ptr = (void **)listmap2vectoraddr((list *)glist);

		if ((parallel->brick = (Pbricks **) Calloc((groups + 1),
                         Pbricks * )) == NULL) {
                free(node_vector_ptr);
                free(parallel);
                break;
        }

		if ((parallel->brick[0] = Calloc((groups + 1) , Pbricks ))
			 == NULL) {
                free(node_vector_ptr);
                free(parallel->brick);
                free(parallel);
                break;
        }

		bptr = parallel->brick[0];
		bptr->max_tail = bptr->begin = bptr->end = last;

		parallel->len_of_bricks = groups + 1;

		for (i = 1; i <= groups; i++) {
			parallel->brick[i] = parallel->brick[i - 1] + 1;
			bptr = parallel->brick[i];
			bptr->max_tail = bptr->begin = bptr->end = last;
		}

		parallel->threads = Calloc( groups+1 ,pthread_t );

        rc = pthread_attr_init(&attr);
        rc = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
   
        size_t stacksize; 
#define THREAD_SIZE (1024*1024*4)
        
        pthread_attr_getstacksize(&attr,&stacksize);
        if (!stacksize) stacksize = 1; 
        while (stacksize < THREAD_SIZE ) stacksize *= 2;
              pthread_attr_setstacksize(&attr, stacksize);

		for (i = 0; i <= groups; i++) {

			bptr = parallel->brick[i];
			bptr->max_tail = last;

			bptr->global_parameters = xFilter;
			bptr->begin = i * stride;
			if (bptr->begin >= last) break;

			bptr->end = (i + 1) * stride;
			if (bptr->end > last) bptr->end = last;

			bptr->input = node_vector_ptr;
			bptr->output = dst_addr;
			bptr->reward = (void *)parallel;
			bptr->pfuncs = funcs;

			while ( pthread_mutex_trylock(&parallel->lock) != 0 );;
			// pthread_create(&parallel->threads[i], NULL, funcs, (void *)bptr);
       
     aFilter->tpool->tasklist.task_add(&aFilter->tpool->tasklist, funcs,(void *) bptr);

			parallel->pcount++;
			pthread_mutex_unlock(&parallel->lock);

		}
		
	if ((sk_func != NULL) && (parallel != NULL))
		tmplist = parallel->shrunken(parallel, timeout);

    free(parallel->threads);
    free(node_vector_ptr);
    free(parallel->brick[0]);
    free(parallel->brick);

	} while (0);
    
	return (parallel);
}
